package org.caw.hbase.explorer.impl;

import java.io.IOException;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.caw.domain.Row;
import org.caw.exception.CawException;
import org.caw.exception.CawNoRecordFoundException;
import org.caw.exception.CawRecordAlreadyExistsException;
import org.caw.hbase.HBaseTablePool;
import org.caw.hbase.explorer.UpdateManager;
import org.caw.util.ConvertUtil;

public class HBaseUpdateManagerImpl implements UpdateManager {

	private HBaseTablePool pool;

	@Override
	public void create(String tableName, String rowkey, Row row, boolean overwrite) throws CawException {
		HTableInterface table = null;
		try {
			table = pool.getTable(tableName);

			// if record exits and overwrite is false
			if (!overwrite && table.exists(ConvertUtil.buildGet(rowkey, null, 1, null))) {
				throw new CawRecordAlreadyExistsException("Record " + rowkey + " already exists in " + tableName);

				// if record doesn't exits and overwrite is true
			} else if (overwrite && !table.exists(ConvertUtil.buildGet(rowkey, null, 1, null))) {
				throw new CawNoRecordFoundException("Record " + rowkey + " doesn't exists in " + tableName);
			} else if (overwrite && table.exists(ConvertUtil.buildGet(rowkey, null, 1, null))) {
				Delete delete = ConvertUtil.buildDelete(rowkey, null, null);
				table.delete(delete);
			}

			Put p = ConvertUtil.convert(row);
			table.put(p);
			table.flushCommits();
		} catch (IOException e) {
			throw new CawException("Failed to create/overwrite record " + rowkey + " in table " + tableName, e);
		} finally {
			if (table != null) {
				pool.putTable(table);
			}
		}

	}

	@Override
	public void update(String tableName, String rowkey, Row row) throws CawException {
		HTableInterface table = null;
		try {
			table = pool.getTable(tableName);

			if (!table.exists(ConvertUtil.buildGet(rowkey, null, 1, null))) {
				throw new CawNoRecordFoundException("Record " + rowkey + " doesn't exists in " + tableName);
			}

			Put p = ConvertUtil.convert(row);
			table.put(p);
			table.flushCommits();
		} catch (IOException e) {
			throw new CawException("Failed to update record " + rowkey + " in table " + tableName, e);
		} finally {
			if (table != null) {
				pool.putTable(table);
			}
		}
	}

	public HBaseTablePool getPool() {
		return pool;
	}

	public void setPool(HBaseTablePool pool) {
		this.pool = pool;
	}

}
